package cn.demoncat.util.zookeeper;

import cn.demoncat.util.exception.ConnRuntimeException;
import cn.demoncat.util.lang.StringUtil;
import cn.demoncat.util.lang.constant.StringConstant;
import cn.demoncat.util.zookeeper.constant.ServerConstant;
import cn.demoncat.util.zookeeper.entity.*;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException.NoChildrenForEphemeralsException;
import org.apache.zookeeper.KeeperException.NoNodeException;
import org.apache.zookeeper.KeeperException.NodeExistsException;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

/**
 * Zookeeper操作工具
 * 
 * @author 延晓磊
 *
 * @since 2018年11月2日
 * 
 * @see <a href="http://curator.apache.org">Curator</a>
 */
@Component("zkUtil")
public class ZkUtil {

	@Resource
	private ZkConnectionUtil zkConnectionUtil;

	// 选主缓存
	private static final Map<String,Boolean> SELECT_MASTER_MAP = new HashMap<>();
	
	private static final Logger LOG = LoggerFactory.getLogger(ZkUtil.class);
	
	/**
	 * 获取服务端信息
	 * 
	 * @param ip
	 * @param port
	 * @return
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月6日
	 */
	public static ZkServerInfo getServer(String ip, int port) {
		ZkServerInfo info = new ZkServerInfo();
		info.setIp(ip);
		info.setPort(port);
		info.setStatus(ServerConstant.Status.OFFLINE);
		// 查询服务状态信息
		ZkServerStat stat = CommandUtil.stat(ip, port);
		if (stat != null) {
			info.setStat(stat);
			// 查询服务运行状态
			if (CommandUtil.ruok(ip, port)) {
				info.setStatus(ServerConstant.Status.ONLINE);
				// 查询服务配置
				info.setConf(CommandUtil.conf(ip, port));
			}else {
				info.setStatus(ServerConstant.Status.EXCEPTION);
			}
		}
		return info;
	}
	
	/**
	 * 获取服务端信息列表
	 * 
	 * @param url	服务端地址列表：多个用,分隔，IP和端口用:分隔
	 * @return
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月6日
	 */
	public static List<ZkServerInfo> getServers(String url) {
		List<ZkServerInfo> servers = new ArrayList<>();
		ZkServerInfo serverInfo;
		String[] host;
		for (String u : url.split(StringConstant.COMMA)) {
			// 服务信息
			host = u.split(StringConstant.COLON);
			serverInfo = ZkUtil.getServer(host[0], Integer.parseInt(host[1]));
			// 转换
			serverInfo.setStatus(ServerConstant.Status.MAP.get(serverInfo.getStatus()));
			if (serverInfo.getStat() != null ) {
				serverInfo.getStat().setMode(ServerConstant.Mode.MAP.get(serverInfo.getStat().getMode()));
			}
			servers.add(serverInfo);
		}
		return servers;
	}
	
	/**
	 * 是否为持久节点
	 * 
	 * @param stat
	 * @return
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月6日
	 */
	public static boolean isPersistent(Stat stat) {
		return stat.getEphemeralOwner() == 0;
	}
	
	/**
	 * 是否为父节点
	 * 
	 * @param stat
	 * @return
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月6日
	 */
	public static boolean isParent(Stat stat) {
		return stat.getNumChildren() > 0;
	}
	
	/**
	 * 连接ZK并执行操作 - 新建连接
	 * 
	 * @param task	回调任务
	 * @return
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月2日
	 */
	public <T>T execute(ZkTask<T> task) {
		// 获取连接
		CuratorFramework conn = zkConnectionUtil.getConnection();
		// 执行操作
		try {
			return task.execute(conn);
		} finally {
			// 关闭连接
			zkConnectionUtil.closeConnection(conn);
		}
	}
	
	/**
	 * ZK分布式锁 - 顺序执行 - execute(ZkTask)内复用连接
	 * 
	 * 场景：A获取锁开始执行 > B未获取锁等待 > A执行完成释放锁 > B获取锁开始执行
	 * 原理：A创建Node成功，执行 > B创建Node失败，等待 > A执行完成，删除Node > B创建Node成功，执行 
	 * 
	 * @param conn		ZK连接
	 * @param task		ZK锁任务：包含 execute,failLock,afterLock
	 * @param lockPath	锁节点路径
	 * @param timeout	等待锁的超时秒数
	 * 
	 * @return	未获取锁(未执行)返回null
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月2日
	 */
	public <T> T lockExecute(CuratorFramework conn, ZkLockTask<T> task, String lockPath, int timeout) {
		T t;
		// 创建进程互斥锁
		InterProcessMutex lock = new InterProcessMutex(conn, lockPath);
		boolean isSuccess = false;
		try {
			// 获取互斥锁：如果锁可用则获取，如果超时未获取则放弃
			boolean acquire;
			try {
				acquire = lock.acquire(timeout, TimeUnit.SECONDS);
			} catch (Exception e) {
				throw new ConnRuntimeException("获取ZK互斥锁失败：lockPath = "+ lockPath, e);
			}
			// 获取成功
			if (acquire) {
				// 执行获取锁的操作
				t = task.execute(conn);
				isSuccess = true;
			}else {
				LOG.warn("获取分布式锁失败： key = {}", lockPath);
				// 执行未获取锁的操作
				t = task.failLock(conn);
			}
		} finally {
			if (lock.isAcquiredInThisProcess()) {
				try {
					// 释放锁
					lock.release();
				} catch (Exception e) {
					LOG.warn("释放互斥锁失败：lockPath = {}", lockPath, e);
				}
				// 执行释放锁后的操作
				task.afterLock(conn, isSuccess);
			}
		}
		return t;
	}
	
	/**
	 * ZK分布式锁 - 顺序执行 - 新建连接
	 * 
	 * 场景：A获取锁开始执行 > B未获取锁等待 > A执行完成释放锁 > B获取锁开始执行
	 * 原理：A创建Node成功，执行 > B创建Node失败，等待 > A执行完成，删除Node > B创建Node成功，执行 
	 * 
	 * @param task		ZK锁任务：包含 execute,failLock,afterLock
	 * @param lockPath	锁节点路径
	 * @param timeout	等待锁的超时秒数
	 * 
	 * @return	未获取锁(未执行)返回null
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月2日
	 */
	public <T> T lockExecute(ZkLockTask<T> task, String lockPath, int timeout) {
		return execute(conn -> lockExecute(conn, task, lockPath, timeout));
	}
	
	/**
	 * ZK分布式锁 - 单次执行 - execute(ZkTask)内复用连接
	 * 
	 * 场景：A获取锁开始执行 > B未获取锁放执行
	 * 原理：A创建Node成功，执行 > B创建Node失败，放弃
	 * 注意：本方法并没有对锁文件进行清除，需要根据情况编写定时器来清除
	 * 
	 * @param conn		ZK连接
	 * @param task		ZK锁任务：包含 execute,failLock
	 * @param lockPath	锁节点路径：父目录链中包含"日期时间"节点，便于通过定时器清除无用的锁文件；完整路径包含id/code等来唯一标识操作
	 * 
	 * @return 未获取锁(未执行)返回null
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月2日
	 */
	public <T> T onceLockExecute(CuratorFramework conn, ZkLockTask<T> task, String lockPath) {
		try {
			// 创建持久节点
			conn.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(lockPath);
			// 创建成功（获取锁），执行任务
			return task.execute(conn);
		} catch (NodeExistsException e) {
			LOG.warn("获取分布式锁失败： key = {}", lockPath);
			// 节点已存在，执行未获取锁的操作
			return task.failLock(conn);
		} catch (Exception e) {
			throw new ConnRuntimeException("创建节点失败：path = " + lockPath, e);
		}
	}
	
	/**
	 * ZK分布式锁 - 单次执行 - 新建连接
	 * 
	 * 场景：A获取锁开始执行 > B未获取锁放执行
	 * 原理：A创建Node成功，执行 > B创建Node失败，放弃
	 * 注意：本方法并没有对锁文件进行清除，需要根据情况编写定时器来清除
	 * 
	 * @param task		ZK锁任务：包含 execute,failLock
	 * @param lockPath	锁节点路径：父目录链中包含"日期时间"节点，便于通过定时器清除无用的锁文件；完整路径包含id/code等来唯一标识操作
	 * 
	 * @return 未获取锁(未执行)返回null
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月2日
	 */
	public <T> T onceLockExecute(ZkLockTask<T> task, String lockPath) {
		return execute(conn -> onceLockExecute(conn, task, lockPath));
	}
	
	/**
	 * 创建节点 - execute(ZkTask)内复用连接
	 * 
	 * @param conn	ZK连接
	 * @param mode	模式：持久节点(永久存在)，持久序列节点(path后缀递增序列)，临时节点(会话结束时自动删除)，临时序列节点
	 * @param path	节点路径
	 * @param data	节点数据
	 * 
	 * @return	节点的完整路径；已存在返回null
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public String create(CuratorFramework conn, CreateMode mode, String path, byte[] data) {
		try {
			// 创建节点：创建必要的父目录，指定模式，指定路径；返回创建的节点的全路径；如果节点已存在则抛异常
			return conn.create().creatingParentsIfNeeded().withMode(mode).forPath(path,data);
		} catch (NodeExistsException e) {
			// 节点已存在
			return null;
		} catch (Exception e) {
			throw new ConnRuntimeException("创建节点失败：path = " + path, e);
		}
	}
	
	/**
	 * 创建节点 - 新建连接
	 * 
	 * @param mode	模式：持久节点(永久存在)，持久序列节点(path后缀递增序列)，临时节点(会话结束时自动删除)，临时序列节点
	 * @param path	节点路径
	 * @param data	节点数据
	 * 
	 * @return	节点的完整路径；已存在返回null
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public String create(CreateMode mode, String path, byte[] data) {
		return execute(conn -> create(conn, mode, path, data));
	}
	
	/**
	 * 创建节点 - execute(ZkTask)内复用连接
	 * 
	 * @param conn	ZK连接
	 * @param mode	模式：持久节点(永久存在)，持久序列节点(path后缀递增序列)，临时节点(会话结束时自动删除)，临时序列节点
	 * @param path	节点路径
	 * @param data	节点数据
	 * 
	 * @return	节点的完整路径；已存在返回null
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public String create(CuratorFramework conn, CreateMode mode, String path, String data) {
		return create(conn, mode, path, StringUtil.getBytes(data));
	}
	/**
	 * 创建节点 - 新建连接
	 * 
	 * @param mode	模式：持久节点(永久存在)，持久序列节点(path后缀递增序列)，临时节点(会话结束时自动删除)，临时序列节点
	 * @param path	节点路径
	 * @param data	节点数据
	 * 
	 * @return	节点的完整路径；已存在返回null
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public String create(CreateMode mode, String path, String data) {
		return create(mode, path, StringUtil.getBytes(data));
	}
	
	/**
	 * 检查节点信息 - execute(ZkTask)内复用连接
	 * 
	 * @param conn	ZK连接
	 * @param path	节点路径
	 * 
	 * @return Stat 节点信息，参考{@link ZkNodeInfo }；不存在返回null
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public Stat check(CuratorFramework conn, String path) {
		try {
			// 检查节点是否存在：返回节点信息，包含节点ID、version、子节点数、创建时间等，参考ZkNodeInfo类
			return conn.checkExists().forPath(path);
		} catch (NoNodeException e) {
			// 节点不存在
			return null;
		} catch (Exception e) {
			throw new ConnRuntimeException("检查节点失败：path = " + path, e);
		}
	}
	
	/**
	 * 检查节点信息 - 新建连接
	 * 
	 * @param path	节点路径
	 * 
	 * @return Stat 节点信息，参考{@link ZkNodeInfo }；不存在返回null
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public Stat check(String path) {
		return execute(conn -> check(conn, path));
	}
	
	/**
	 * 获取节点数据 - execute(ZkTask)内复用连接
	 * 
	 * @param conn	ZK连接
	 * @param path	节点路径
	 * 
	 * @return 节点数据；不存在返回null
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public byte[] get(CuratorFramework conn, String path) {
		try {
			//获取节点的数据
			return conn.getData().forPath(path);
		} catch (NoNodeException e) {
			// 节点不存在
			return null;	
		} catch (Exception e) {
			throw new ConnRuntimeException("获取节点失败：path = " + path, e);
		}
	}
	
	/**
	 * 获取节点数据 - 新建连接
	 * 
	 * @param path	节点路径
	 * 
	 * @return 节点数据；不存在返回null
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public byte[] get(String path) {
		return execute(conn -> get(conn, path));
	}
	
	/**
	 * 获取节点数据 - execute(ZkTask)内复用连接
	 * 
	 * @param conn	ZK连接
	 * @param path	节点路径
	 * 
	 * @return 节点数据；不存在返回null
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public String getData(CuratorFramework conn, String path) {
		return StringUtil.getStr(get(conn, path));
	}
	
	/**
	 * 获取节点数据 - 新建连接
	 * 
	 * @param path	节点路径
	 * 
	 * @return 节点数据；不存在返回null
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public String getData(String path) {
		return StringUtil.getStr(get(path));
	}
	
	/**
	 * 修改节点数据 - execute(ZkTask)内复用连接
	 * 
	 * @param conn	ZK连接
	 * @param path	节点路径
	 * @param data	节点数据
	 * 
	 * @return Stat 节点信息，参考{@link ZkNodeInfo }；不存在返回null
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public Stat set(CuratorFramework conn, String path, byte[] data) {
		try {
			//修改节点的数据：返回节点信息
			return conn.setData().forPath(path, data);
		} catch (NoNodeException e) {
			// 节点不存在
			return null;
		} catch (Exception e) {
			throw new ConnRuntimeException("修改节点失败：path = " + path, e);
		}
	}
	
	/**
	 * 修改节点数据 - 新建连接
	 * 
	 * @param path	节点路径
	 * @param data	节点数据
	 * 
	 * @return Stat 节点信息，参考{@link ZkNodeInfo }；不存在返回null
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public Stat set(String path, byte[] data) {
		return execute(conn -> set(conn, path, data));
	}
	
	/**
	 * 修改节点数据 - execute(ZkTask)内复用连接
	 * 
	 * @param conn	ZK连接
	 * @param path	节点路径
	 * @param data	节点数据
	 * 
	 * @return Stat 节点信息，参考{@link ZkNodeInfo }；不存在返回null
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public Stat setData(CuratorFramework conn, String path, String data) {
		return set(conn, path, StringUtil.getBytes(data));
	}
	
	/**
	 * 修改节点数据 - 新建连接
	 * 
	 * @param path	节点路径
	 * @param data	节点数据
	 * 
	 * @return Stat 节点信息，参考{@link ZkNodeInfo }；不存在返回null
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public Stat setData(String path, String data) {
		return set(path, StringUtil.getBytes(data));
	}
	
	/**
	 * 删除节点 - execute(ZkTask)内复用连接
	 * 
	 * @param conn	ZK连接
	 * @param path	节点路径
	 * 
	 * @return 删除成功返回true；不存在返回false
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public Boolean del(CuratorFramework conn, String path) {
		try {
			//删除节点：强制删除，级联子节点
			conn.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);
			return true;
		} catch (NoNodeException e) {
			// 节点不存在
			return false;
		} catch (Exception e) {
			throw new ConnRuntimeException("删除节点失败：path = " + path, e);
		}
	}
	
	/**
	 * 删除节点 - 新建连接
	 * 
	 * @param path	节点路径
	 * 
	 * @return 删除成功返回true；不存在返回false
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public Boolean del(String path) {
		return execute(conn -> del(conn, path));
	}
	
	/**
	 * 获取子节点集 - execute(ZkTask)内复用连接
	 * 
	 * @param conn	ZK连接
	 * @param path	节点路径
	 * 
	 * @return 子节点集(节点名称，非完整路径名)；没有子节点返回[]；不存在返回null
	 * @throws ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public List<String> child(CuratorFramework conn, String path) {
		try {
			//获取子节点集
			return conn.getChildren().forPath(path);
		} catch (NoNodeException | NoChildrenForEphemeralsException e) {
			// 节点不存在，没有子节点
			return null;
		} catch (Exception e) {
			throw new ConnRuntimeException("获取子节点失败：path = " + path, e);
		}
	}
	
	/**
	 * 获取子节点集 - 新建连接
	 * 
	 * @param path	节点路径
	 * 
	 * @return 子节点集(节点名称，非完整路径名)；没有子节点返回[]；不存在返回null
	 * @throws ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public List<String> child(String path) {
		return execute(conn -> child(conn,path));
	}
	
	/**
	 * 监听节点 - 创建、更新、删除 - execute(ZkTask)内复用连接
	 * 
	 * 注：默认无限次监听，主动停止监听可以在 task 中调用 CloseUtil.close(watcher)
	 * 
	 * @param conn	ZK连接
	 * @param path	节点路径
	 * @param task	事件回调
	 * @param executor	线程池（异步监听）
	 * 
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public Boolean watchNode(CuratorFramework conn, String path, ZkWatchTask task,Executor executor) {
		// 创建节点Watcher：监听节点的创建、更新、删除；原理：将节点的数据缓存在本地进行比对
		NodeCache watcher = new NodeCache(conn, path);
		// 添加监听器
		// 节点变化
		watcher.getListenable().addListener(() -> {
			// 执行回调
			task.watch(watcher);
		},executor);
		// 启动Watcher，保存节点的初始视图
		try {
			watcher.start(true);
			return true;
		} catch (Exception e) {
			throw new ConnRuntimeException("监听节点失败：path = " + path, e);
		}
	}
	
	/**
	 * 监听节点 - 创建、更新、删除 - 新建连接
	 * 
	 * 注：默认无限次监听，主动停止监听可以在 task 中调用 CloseUtil.close(watcher)
	 * 
	 * @param path	节点路径
	 * @param task	事件回调
	 * @param executor	线程池（异步监听）
	 * 
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public Boolean watchNode(String path, ZkWatchTask task,Executor executor) {
		return execute(conn -> watchNode(conn, path, task, executor));
	}
	
	/**
	 * 监听子节点 - 创建、更新、删除 - execute(ZkTask)内复用连接
	 * 
	 * 注：默认无限次监听，主动停止监听可以在 task 中调用 CloseUtil.close(watcher)
	 * 
	 * @param conn	ZK连接
	 * @param path	节点路径
	 * @param task	事件回调
	 * @param executor	线程池（异步监听）
	 * 
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public Boolean watchChild(CuratorFramework conn, String path, ZkWatchTask task,Executor executor) {
		// 创建子节点Watcher：监听子节点的创建、更新、删除、连接变更；原理：将子节点列表及其数据缓存在本地进行比对
		PathChildrenCache watcher = new PathChildrenCache(conn, path, true);
		// 添加监听器
		watcher.getListenable().addListener((client, event) -> {
			// 筛选节点变更状态
			if (PathChildrenCacheEvent.Type.CHILD_ADDED.equals(event.getType()) || PathChildrenCacheEvent.Type.CHILD_UPDATED.equals(event.getType()) || PathChildrenCacheEvent.Type.CHILD_REMOVED.equals(event.getType())) {
				// 执行回调
				task.watch(watcher, event);
			}
			// 忽略连接变更状态
		},executor);
		// 启动Watcher，保存节点的初始视图
		try {
			watcher.start(StartMode.BUILD_INITIAL_CACHE);
			return true;
		} catch (Exception e) {
			throw new ConnRuntimeException("监听子节点失败：path = " + path, e);
		}
	}
 
	/**
	 * 监听子节点 - 创建、更新、删除 - 新建连接
	 * 
	 * 注：默认无限次监听，主动停止监听可以在 task 中调用 CloseUtil.close(watcher)
	 * 
	 * @param path	节点路径
	 * @param task	事件回调
	 * @param executor	线程池（异步监听）
	 * 
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public Boolean watchChild(String path, ZkWatchTask task,Executor executor) {
		return execute(conn -> watchChild(conn, path, task, executor));
	}
	
	/**
	 * 监听树节点 - 创建、更新、删除 - execute(ZkTask)内复用连接
	 * 
	 * 注：默认无限次监听，主动停止监听可以在 task 中调用 CloseUtil.close(watcher)
	 * 
	 * @param conn	ZK连接
	 * @param path	节点路径
	 * @param task	事件回调
	 * @param executor	线程池（异步监听）
	 * 
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public Boolean watchTree(CuratorFramework conn, String path, ZkWatchTask task,Executor executor) {
		// 创建树节点Watcher：监听所有后代子节点的创建、更新、删除；原理：将节点树及其数据缓存在本地进行比对
		TreeCache watcher = new TreeCache(conn, path);
		// 添加监听器
		watcher.getListenable().addListener((client, event) -> {
			// 筛选节点变更状态
			if (TreeCacheEvent.Type.NODE_ADDED.equals(event.getType()) || TreeCacheEvent.Type.NODE_UPDATED.equals(event.getType()) || TreeCacheEvent.Type.NODE_REMOVED.equals(event.getType())) {
				// 执行回调
				task.watch(watcher, event);
			}
			// 忽略连接变更状态
		},executor);
		// 启动Watcher
		try {
			watcher.start();
			return true;
		} catch (Exception e) {
			throw new ConnRuntimeException("监听树节点失败：path = " + path, e);
		}
	}
	
	/**
	 * 监听树节点 - 创建、更新、删除 - 新建连接
	 * 
	 * 注：默认无限次监听，主动停止监听可以在 task 中调用 CloseUtil.close(watcher)
	 * 
	 * @param path	节点路径
	 * @param task	事件回调
	 * @param executor	线程池（异步监听）
	 * 
	 * @throws 	ConnRuntimeException 操作失败、连接失败时抛异常
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月3日
	 */
	public Boolean watchTree(String path, ZkWatchTask task,Executor executor) {
		return execute(conn -> watchTree(conn, path, task, executor));
	}

	/**
	 * 参与选主
	 *
	 * @param path
	 *
	 * @author 延晓磊
	 * @since 2019年08月29日
	 */
	public void selectMaster(String path){
		// 获取连接
		CuratorFramework conn = zkConnectionUtil.getConnection();
		// 创建选择器
		LeaderSelector selector = new LeaderSelector(conn, path,
				new LeaderSelectorListenerAdapter() {
					@Override
					public void takeLeadership(CuratorFramework client) {
						SELECT_MASTER_MAP.put(path, true);
					}
				});
		// 开启选主
		selector.autoRequeue();
		selector.start();
	}

	/**
	 * 校验是否为主
	 *
	 * @param path
	 * @return
	 *
	 * @author 延晓磊
	 * @since 2019年08月29日
	 */
	public static boolean checkMaster(String path){
		Boolean bool = SELECT_MASTER_MAP.get(path);
		return bool != null && bool;
	}

}
